package com.bw.sparkstreaming.job5
import org.apache.log4j.{Level, Logger}
import org.apache.spark.streaming._
import org.apache.spark.{SparkConf, SparkContext}
/**
 * updateStateBykey 官网上能看到
 * mapWithState 官方博客上面，有，而且说测试过性能更好
 *               并且在Spark源码的例子包里面
 *               演示了这个算子是如何使用的？
 */
object MapWithStateAPITest {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.ERROR)

    val sparkConf = new SparkConf().setAppName("NetworkWordCount").setMaster("local[2]")
    val sc = new SparkContext(sparkConf)
    val ssc = new StreamingContext(sc, Seconds(2))

    ssc.checkpoint("E:\\project_workspace\\bw-sparkstreaming\\mapwithstatedir")
    val lines = ssc.socketTextStream("localhost", 9999)
    val words = lines.flatMap(_.split(","))
    val wordsDStream = words.map(x => (x, 1))

    //定义初始值
    val initialRDD = sc.parallelize(List(("flink", 100L), ("sparkstreaming", 32L)))

    // currentBatchTime : 表示当前的Batch的时间
    // key: 表示需要更新状态的key
    // value: 表示当前batch的对应的key的对应的值
    // currentState: 对应key的当前的状态

    /**
      *
      * hadoop,1
      * hadoop,1
      * hadoop,1
      *
      * {hadoop,(1,1,1)  => 3}
     *
     *  hadoop,3
     *
     *  hadoop,10
     *
      * key:hadoop  当前的key
      * value:3  当前的key出现的次数
      *lastState： 当前的这个key的历史的状态
      *
      * hadoop:3
      *
      * hadoop,10
     *
     *  hadoop,13
     *
     */

    val stateSpec =StateSpec.function((currentBatchTime: Time, key: String, value: Option[Int], lastState: State[Long]) => {

      val sum = value.getOrElse(0).toLong + lastState.getOption.getOrElse(0L)

      val output = (key, sum)
      //如果你的数据没有超时
      if (!lastState.isTimingOut()) {
        lastState.update(sum)
      }
      //最后一行代码是返回值
      Some(output)
    }).initialState(initialRDD)
      .numPartitions(3).timeout(Seconds(15))
    //timeout: 当一个key超过这个时间没有接收到数据的时候，这个key以及对应的状态会被移除掉

    /**
      * reduceByKey
     *
      * udpateStateByKey
      * mapWithState // bykey -> 顺带就完成了合并的操作
      */
    val result = wordsDStream.mapWithState(stateSpec)

   // result.print()

    result.stateSnapshots().print()

    //启动Streaming处理流
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()
  }

}